package com.shujia.sql


import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}


object Demo3DSLApi {
  def main(args: Array[String]): Unit = {
    //创建SparkSession对象
    val sparkSession: SparkSession = SparkSession.builder()
      .config("spark.sql.shuffle.partitions","1")
      .master("local")
      .appName("DSL语法风格编写spark sql")
      .getOrCreate()

    /**
     * 如果要想使用DSL语法编写spark sql的话，需要导入两个隐式转换
     */
    //将sql中的函数，封装成spark程序中的一个个的函数直接调用，以传参的方式调用
    import org.apache.spark.sql.functions._
    //主要作用是，将来可以在调用的函数中，使用$函数，将列名字符串类型转成一个ColumnName类型
    //而ColumnName是继承自Column类的
    import sparkSession.implicits._

    /**
     * 读取json数据文件，转成DF
     * 读取json数据的时候，是不需要指定表结构，可以自动根据json的键值来构建DataFrame
     */
    //老版本的读取json数据的方式
    //    val df1: DataFrame = sparkSession.read
    //      .format("json")
    //      .load("spark/data/students.json")

    val df1: DataFrame = sparkSession.read
      .json("spark/data/students.json")

    //默认显示20行数据
    //    df1.show()
    //传入要查看的行数
    //    df1.show(100)
    //传入第二个参数，观察的更加详细,默认每一列只会保留20个字符
    //    df1.show(10,truncate = false)

    /**
     * DSL语法的第一个函数 select
     * 类似于纯sql语法中的select关键字，传入要查询的列
     */
    //select name,clazz from xxx;
    //        df1.select("name","age").show()
    //    df1.select($"name", $"age").show()
    //查询每个学生的姓名，原本的年龄，年龄+1
    //    df1.select("name", "age").show()
    /**
     * 与select功能差不多的查询函数
     * 如果要以传字符串的形式给到select的话，并且还想对列进行表达式处理的话，可以使用selectExpr函数
     */
    //    df1.selectExpr("name", "age", "age+1 as new_age").show()
    //如果想要使用select函数查询的时候对列做操作的话，可以使用$函数将列变成一个对象
    //    df1.select($"name", $"age", $"age" + 1 as "new_age").show()

    /**
     * DSL语法函数：where
     *
     * === : 类似于sql中的= 等于某个值
     * =!= : 类似于sql中的!=或者<>  不等于某个值
     */
    //    df1.where("gender='男'").show()
    //    df1.where("gender='男' and substring(clazz,0,2)='理科'").show()
    //建议使用隐式转换中的功能进行处理过滤
    //过滤出男生且理科的
    //    df1.where($"gender" === "男" and substring($"clazz",0,2) === "理科").show()
    //过滤出女生且理科的
    //    df1.where($"gender" =!= "男" and substring($"clazz",0,2) === "理科").show()

    /**
     * DSL语法函数：groupBy
     *
     * 非分组字段是无法出现在select查询语句中的
     */
    //查询每个班级的人数
    //    df1.groupBy("clazz")
    //      .agg(count("clazz") as "counts")
    //      .show()

    /**
     * DSL语法函数：orderBy
     */
//    df1.groupBy("clazz")
//      .agg(count("clazz") as "counts")
//      .orderBy($"counts".desc)
//      .show(3)

    /**
     * DSL语法函数: join
     */
    val df2: DataFrame = sparkSession.read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,subject_id STRING,score INT")
      .load("spark/data/score.txt")

    // 关联的字段名不一样的情况
//    df2.join(df1,$"id" === $"sid","inner")
//      .select("id","name","age","gender","clazz","subject_id","score")
//      .show(10)
    // 关联的字段名一样的情况
//    df2.join(df1,"id")
//      .select("id","name","age","gender","clazz","subject_id","score")
//      .show(10)

    //如果关联的字段名一样且想使用其他连接方式的话,可以将字段名字用Seq()传入,同时可以传连接方式
//    df2.join(df1, Seq("id"),"left")
//          .select("id","name","age","gender","clazz","subject_id","score")
//          .show(10)

    /**
     * DSL语法函数: 开窗
     * 无论是在纯sql中还是在DSL语法中,开窗是不会改变原表条数
     */
    //计算每个班级总分前3的学生
    //纯sql的方式实现
//    df1.createOrReplaceTempView("students")
//    df2.createOrReplaceTempView("scores")
//    sparkSession.sql(
//      """
//        |select
//        |*
//        |from
//        |(
//        |select t1.id,
//        |t2.name,
//        |t2.clazz,
//        |t1.sumScore,
//        |row_number() over(partition by t2.clazz order by t1.sumScore desc) as rn
//        |from
//        |(
//        | select id,
//        |        sum(score) as sumScore
//        | from
//        |   scores
//        | group by id) t1
//        |join
//        | students t2
//        |on(t1.id=t2.id)) tt1 where tt1.rn<=3
//        |""".stripMargin).show()

    //DSL语法实现
//    df2.groupBy("id") //根据学号分组
//      .agg(sum("score") as "sumScore") // 计算每个人总分
//      .join(df1,"id") // 与学生信息表关联,得到班级列
//      .select($"id",$"name",$"clazz",$"sumScore",row_number() over Window.partitionBy("clazz").orderBy($"sumScore".desc) as "rn")
//      .where($"rn" <= 3)
////      .repartition(1)
//      .write
//      .format("csv")
//      .mode(SaveMode.Overwrite)
//      .save("spark/data/sqlout3")

    df2.groupBy("id") //根据学号分组
      .agg(sum("score") as "sumScore") // 计算每个人总分
      .join(df1, "id") // 与学生信息表关联,得到班级列
      .withColumn("rn",row_number() over Window.partitionBy("clazz").orderBy($"sumScore".desc)) // 新增一列
//      .select($"id", $"name", $"clazz", $"sumScore", row_number() over Window.partitionBy("clazz").orderBy($"sumScore".desc) as "rn")
      .show()
  }
}
